package com.margo.project.curator.config;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.curator.RetryPolicy;
import org.apache.curator.ensemble.EnsembleProvider;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;

//import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 
 * @author fangxingbang
 *
 */
@Configuration
//@Configurable
public class MargoZookeeperConfiguration {
	
	private static final Log log = LogFactory.getLog(MargoZookeeperConfiguration.class);
	
	/**
	 *  监听
	 * @param curatorFramework
	 * @return
	 * @throws Exception
	 */
	@Bean
	public NodeCache nodeCache(CuratorFramework curatorFramework) throws Exception {
		NodeCache nodeCache= new NodeCache(curatorFramework, "/margo");
		nodeCache.start();
		return nodeCache;
	}
	
	/**
	 * 	分布式锁
	 * @param curatorFramework
	 * @return
	 */
	@Bean
	public InterProcessMutex interProcessMutex(CuratorFramework curatorFramework) {
		return new InterProcessMutex(curatorFramework,"/reentrantlock");
	}
	
	@Autowired(required = false)
	private EnsembleProvider ensembleProvider;

	@Bean
	@ConditionalOnMissingBean
	public ZookeeperProperties zookeeperProperties() {
		return new ZookeeperProperties();
	}

	@Bean(destroyMethod = "close")
	@ConditionalOnMissingBean
	public CuratorFramework curatorFramework(RetryPolicy retryPolicy,
			ZookeeperProperties properties) throws Exception {
		CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder();
		if (this.ensembleProvider != null) {
			builder.ensembleProvider(this.ensembleProvider);
		}
		else {
			builder.connectString(properties.getConnectString());
		}
		CuratorFramework curator = builder.retryPolicy(retryPolicy).build();
		curator.start();
		log.trace("blocking until connected to zookeeper for "
				+ properties.getBlockUntilConnectedWait()
				+ properties.getBlockUntilConnectedUnit());
		curator.blockUntilConnected(properties.getBlockUntilConnectedWait(),
				properties.getBlockUntilConnectedUnit());
		log.trace("connected to zookeeper");
		return curator;
	}

	@Bean
	@ConditionalOnMissingBean
	public RetryPolicy exponentialBackoffRetry(ZookeeperProperties properties) {
		return new ExponentialBackoffRetry(properties.getBaseSleepTimeMs(),
				properties.getMaxRetries(), properties.getMaxSleepMs());
	}

}
